/*
 * Copyright 2011-2012 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.sabayframework.activity.impl;

import java.util.concurrent.locks.ReentrantReadWriteLock;

import org.sabayframework.activity.Activity;
import org.sabayframework.activity.ActivityContext;
import org.sabayframework.activity.impl.BlockingActivity;
import org.sabayframework.channel.Lock;

/**
 * An activity to access a shared resource.  
 * It is recommended that read and write operations were CPU-bound only.
 * @author Alex Antonau
 * */
public abstract class ReadWriteActivity<I, O, P>  {

	final private ReentrantReadWriteLock lock;
	final private ReadActivity readActivity;
	final private WriteActivity writeActivity;

	public ReadWriteActivity() {
		super();
		this.lock = new ReentrantReadWriteLock();
		this.readActivity = new ReadActivity();
		this.writeActivity = new WriteActivity();
	}
	
	private final class ReadActivity extends BlockingActivity<I, O>{
		P param;
		@Override
		public boolean isReady(ActivityContext<I, O> ctx) throws Exception {
			return super.isReady(ctx) && !lock.isWriteLocked();
		}
		@Override
		public void execute(ActivityContext<I, O> ctx, Lock<I> inLock, Lock<O> outLock) throws Exception {
			if(lock.readLock().tryLock()){
				try{
					read(ctx, inLock, outLock, param);
				} finally{
					lock.readLock().unlock();
				}
			}else{
				inLock.cancel();
				outLock.cancel();
			}
		}
	}
	private final class WriteActivity extends BlockingActivity<I, O>{
		P param;
		@Override
		public boolean isReady(ActivityContext<I, O> ctx) throws Exception {
			return super.isReady(ctx) && !lock.isWriteLocked() && lock.getReadLockCount() == 0;
		}
		@Override
		public void execute(ActivityContext<I, O> ctx, Lock<I> inLock, Lock<O> outLock) throws Exception {
			if(lock.writeLock().tryLock()){
				try{
					write(ctx, inLock, outLock, param);
				} finally{
					lock.writeLock().unlock();
				}
			}else{
				inLock.cancel();
				outLock.cancel();
			}
		}
	}
	/**
	 * Returns a parameterized read activity to submit into execution queue.
	 * */
	public final Activity<I, O> getReadActivity(P param) {
		ReadActivity ra = new ReadActivity();
		ra.param = param;
		return ra;
	}

	/**
	 * Returns a parameterized write activity to submit into execution queue.
	 * */
	public final Activity<I, O> getWriteActivity(P param) {
		WriteActivity wa = new WriteActivity();
		wa.param = param;
		return wa;
	}
	/**
	 * Returns a read activity to submit into execution queue.
	 * */
	public final Activity<I, O> getReadActivity() {
		return readActivity;
	}

	/**
	 * Returns a write activity to submit into execution queue.
	 * */
	public final Activity<I, O> getWriteActivity() {
		return writeActivity;
	}
	
	/**
	 * Read data from underlying shared resource. This method can be called concurrently.
	 * @param ctx activity context
	 * @param inLock locked input data
	 * @param outLock locked output data
	 * @param param a parameter for activity
	 * */
	public abstract void read(ActivityContext<I, O> ctx, Lock<I> inLock, Lock<O> outLock, P param) throws Exception;
	/**
	 * Write data to underlying shared resource. This method can be called only after an exclusive lock is obtained. 
	 * @param ctx activity context
	 * @param inLock locked input data
	 * @param outLock locked output data
	 * @param param a parameter for activity
	 * */
	public abstract void write(ActivityContext<I, O> ctx, Lock<I> inLock, Lock<O> outLock, P param) throws Exception;

}
